package org.groupg.project;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;

public class DisruptorDAGExample {

    // 定义事件类
    static class TaskEvent {
        private int taskId;

        public int getTaskId() {
            return taskId;
        }

        public void setTaskId(int taskId) {
            this.taskId = taskId;
        }
    }

    // 定义事件处理器 A
    static class TaskHandlerA implements com.lmax.disruptor.EventHandler<TaskEvent> {
        @Override
        public void onEvent(TaskEvent event, long sequence, boolean endOfBatch) {
            System.out.println("Handler A processing task: " + event.getTaskId());
            // 模拟处理时间
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    // 定义事件处理器 B
    static class TaskHandlerB implements com.lmax.disruptor.EventHandler<TaskEvent> {
        @Override
        public void onEvent(TaskEvent event, long sequence, boolean endOfBatch) {
            System.out.println("Handler B processing task: " + event.getTaskId());
            // 模拟处理时间
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    // 定义事件处理器 C
    static class TaskHandlerC implements com.lmax.disruptor.EventHandler<TaskEvent> {
        @Override
        public void onEvent(TaskEvent event, long sequence, boolean endOfBatch) {
            System.out.println("Handler C processing task: " + event.getTaskId());
            // 模拟处理时间
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        // 定义 Disruptor 的缓冲区大小
        int bufferSize = 1024;

        // 创建 Disruptor
        Disruptor<TaskEvent> disruptor = new Disruptor<>(
                TaskEvent::new, bufferSize, DaemonThreadFactory.INSTANCE);

        // 创建事件处理器
        TaskHandlerA handlerA = new TaskHandlerA();
        TaskHandlerB handlerB = new TaskHandlerB();
        TaskHandlerC handlerC = new TaskHandlerC();

        // 设置事件处理器的依赖关系
        // 假设依赖关系为：A -> B -> C
        disruptor.handleEventsWith(handlerA).then(handlerB).then(handlerC);

        // 启动 Disruptor
        RingBuffer<TaskEvent> ringBuffer = disruptor.start();

        // 发布事件
        for (int i = 0; i < 10; i++) {
            long sequence = ringBuffer.next();
            try {
                TaskEvent event = ringBuffer.get(sequence);
                event.setTaskId(i);
            } finally {
                ringBuffer.publish(sequence);
            }
            Thread.sleep(100); // 模拟事件发布的间隔
        }

        // 关闭 Disruptor
        disruptor.shutdown();
    }
}